package com.ypj.rpc.transport.netty.client;

import com.ypj.rpc.codec.CommonDecoder;
import com.ypj.rpc.codec.CommonEncoder;
import com.ypj.rpc.entity.RpcRequest;
import com.ypj.rpc.entity.RpcResponse;
import com.ypj.rpc.enumeration.RpcError;
import com.ypj.rpc.exception.RpcException;
import com.ypj.rpc.factory.SingletonFactory;
import com.ypj.rpc.loaderbalancer.LoaderBalancer;
import com.ypj.rpc.loaderbalancer.RandomLoadBalancer;
import com.ypj.rpc.registry.NacosServiceDiscovery;
import com.ypj.rpc.registry.ServiceDiscovery;
import com.ypj.rpc.serializer.CommonSerializer;
import com.ypj.rpc.serializer.KryoSerializer;
import com.ypj.rpc.transport.RpcClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

@Slf4j
public class NettyClient implements RpcClient {

    private final ServiceDiscovery serviceDiscovery;
    private final Bootstrap bootstrap;
    private final EventLoopGroup group;
    private CommonSerializer serializer;
    private final ChannelProvider channelProvider;

    private final UnprocessedRequests unprocessedRequests;

    public NettyClient() {
        this(new KryoSerializer(),new RandomLoadBalancer());
    }

    public NettyClient(LoaderBalancer balancer){
        this(new KryoSerializer(),balancer);
    }

    public NettyClient(CommonSerializer serializer, LoaderBalancer balancer) {
        group= new NioEventLoopGroup();
        bootstrap=new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                .option(ChannelOption.SO_KEEPALIVE,true)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new IdleStateHandler(0,5,0,TimeUnit.SECONDS));
                        pipeline.addLast(new CommonEncoder(serializer));
                        pipeline.addLast(new CommonDecoder());
                        pipeline.addLast(new NettyClientHandler());
                    }
                });
        this.serializer = serializer;
        this.serviceDiscovery=new NacosServiceDiscovery(balancer);
        this.channelProvider=SingletonFactory.getInstance(ChannelProvider.class);
        this.unprocessedRequests= SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    /**
     * 将返回值类型改为CompletableFuture类型实现异步，避免执行耗时任务时阻塞太久
     * @param rpcRequest
     * @return
     */
    @Override
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        if (serializer==null){
            log.error("未设置序列化器！");
            throw new RpcException(RpcError.SERIALIZER_NOT_FOUND.getMessage());
        }

        CompletableFuture<RpcResponse> resultFuture=new CompletableFuture<>();
        try {
            InetSocketAddress inetSocketAddress = serviceDiscovery.lookup(rpcRequest.getInterfaceName());
            Channel channel=getChannel(inetSocketAddress);
            if (channel.isActive()){
                unprocessedRequests.put(rpcRequest.getRequestId(),resultFuture);
                channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener)future -> {
                    if (future.isSuccess()){
                        log.info("客户端发送消息：{}",rpcRequest.toString());
                    }else{
                        future.channel().close();
                        resultFuture.completeExceptionally(future.cause());
                        log.error("发送消息时有错误发生：",future.cause());
                    }
                });
            }else{
                channelProvider.remove(inetSocketAddress);
                group.shutdownGracefully();
                return null;
            }
        } catch (InterruptedException | ExecutionException e) {
            unprocessedRequests.remove(rpcRequest.getRequestId());
            log.error("发送消息时有错误发生：",e);
            Thread.currentThread().interrupt();
        }
        return resultFuture;
    }



    /**
     * 客户端发起connect请求，并获取channel保存为静态变量
     * @param inetSocketAddress 连接地址
     */
    private Channel getChannel(InetSocketAddress inetSocketAddress) throws ExecutionException, InterruptedException {
        Channel channel = channelProvider.get(inetSocketAddress);
        if (channel==null){
            channel=doConnect(inetSocketAddress);
            channelProvider.set(inetSocketAddress,channel);
        }
        return channel;
    }

    private Channel doConnect(InetSocketAddress inetSocketAddress) throws ExecutionException, InterruptedException {
            CompletableFuture<Channel> completableFuture=new CompletableFuture<>();
            bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener)future -> {
                if (future.isSuccess()){
                    log.info("客户端连接成功！");
                    completableFuture.complete(future.channel());
                }else{
                    throw new IllegalStateException();
                }
            });
            return completableFuture.get();
    }
}
